package cn.pengpeng.cmcc.app

import java.text.SimpleDateFormat

import cn.pengpeng.cmcc.util.AppParams
import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 移动运营实时监控平台 -- Monitor
  * Created by root on 2018/9/25.
  */
object BootStrapApp {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("BootStrapApp")
      .setMaster("local[*]")
      //序列化方式
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      //开启压缩
      .set("spark.rdd.compress", "true")
      //最大拉取数据
      .set("spark.streaming.kafka.maxRatePerPartition", "1000") //batchSize = partitionNum * 1000 * batchTime

      .set("spark.streaming.stopGracefullyOnShutdown", "true")

    val ssc: StreamingContext = new StreamingContext(conf,Seconds(2))

    //从数据库中拉取偏移量

    //拉取数据
    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](
        AppParams.topics,
        AppParams.kafkaParams,
        Map[TopicPartition, Long]()
    ))
    //输出数据
    //上传
    stream.foreachRDD(x=>println(x+"-----------"))

    stream.foreachRDD(
      rdd =>{
        // baseRdd 使用cache来提高效率
        val baseData: RDD[JSONObject] = rdd
          // ConsumerRecord => JSONObject
          .map(cr => JSON.parseObject(cr.value()))
          // 过滤出充值通知日志
          .filter(obj => obj.getString("serviceName").equalsIgnoreCase("equalsIgnoreCase"))
          .cache()

        val totalSucc: RDD[(String, Int)] = baseData.map(obj => {
          val reqId: String = obj.getString("requestId")
          // 获取当前日期
          val day: String = reqId.substring(0, 8)

          //取出该条充值是否成功的标志
          val result: String = obj.getString("bussinessRst")

          val flag = if (result.equals("0000")) 1 else 0
          (day, flag)

        }).reduceByKey(_ + _)

        // wordcount => (word,1) ==> reduceByKey()
        //每天的充值成功订单  =>(20180503,0) => reudceByKey(_+_)

        val key: RDD[(String, Double)] = baseData.map(obj => {
          val reqId: String = obj.getString("requestId")
          //获取当前日期
          val day = reqId.substring(0, 8)
          //取出该条充值是否成功的标志
          val result: String = obj.getString("bussinessRst")
          //金额
          val fee: Double = if (result.equals("0000")) 1 else 0

          (day, fee)
        }).reduceByKey(_ + _)

        //总订单量
        val count: Long = baseData.count()

        val totalTime: RDD[(String, Long)] = baseData.map(obj => {
          val reqId: String = obj.getString("requestId")
          //取日期
          val day: String = reqId.substring(0, 8)

          //取出该条充值是否成功的标志
          val result: String = obj.getString("bussinessRst")
          val endTime: String = obj.getString("receiveNotifyTime")
          val starTime: String = reqId.substring(0, 17)

          val format: SimpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmssSSS")
          val cost = if (result.equals("0000")) format.parse(endTime).getTime - format.parse(starTime).getTime else 0

          (day, cost)
        }).reduceByKey(_ + _)

        // 保存结果
        totalSucc.foreach(println(_))

      }
    )

    // 启动程序
    ssc.start()
    ssc.awaitTermination()



  }

}
